package com.atguigu.app.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.bean.TableProcess;
import com.atguigu.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.*;

/**
 * @className: TableProcessFunction
 * @author: LinCong
 * @description:
 * @date: 2023/1/11 20:47
 * @version: 1.0
 */
public class TableProcessFunction extends BroadcastProcessFunction<JSONObject, String, JSONObject> {

    private Connection connection;
    private MapStateDescriptor<String, TableProcess> mapStateDescriptor;

    public TableProcessFunction(MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.mapStateDescriptor = mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        connection = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }

    //    {"before":null,"after":{"source_table":"aa","sink_table":"bb","sink_columns":"cc","sink_pk":"id","sink_extend":"xxx"},"source":{"version":"1.5.4.Final","connector":"mysql","name":"mysql_binlog_source","ts_ms":1673452459907,"snapshot":"false","db":"gmall_config","sequence":null,"table":"table_process","server_id":0,"gtid":null,"file":"","pos":0,"row":0,"thread":null,"query":null},"op":"r","ts_ms":1673452459912,"transaction":null}
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
//        1、获取并解析mySqlSourceDS数据
        JSONObject jsonObject = JSON.parseObject(value);
        String after = jsonObject.getString("after");
        TableProcess tableProcess = JSON.parseObject(after, TableProcess.class);
//        2、校验并建表
        if (tableProcess != null ) {
            String sinkTable = tableProcess.getSinkTable();
//            {"database":"gmall_config","table":"table_process","type":"insert","ts":1673443606,"xid":149350,"commit":true,"data":{"source_table":"","sink_table":null,"sink_columns":null,"sink_pk":null,"sink_extend":null}}
            if (sinkTable!=null){
                checkTable(tableProcess.getSinkTable(), tableProcess.getSinkColumns(), tableProcess.getSinkPk(), tableProcess.getSinkExtend());
            }
        }
//        3、写入状态，广播出去(他的数据会广播到所有并行度)
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        if (tableProcess == null) {
            broadcastState.put(null, null);
        } else {
            broadcastState.put(tableProcess.getSourceTable(), tableProcess);
        }

    }

    /**
     * 校验并建表：create table if not exits db.tn(id varchar primary key,bb varchar,cc varchar) xxx
     *
     * @param sinkTable   phoenix表名
     * @param sinkColumns phoenix表字段
     * @param sinkPk      phoenix表主键
     * @param sinkExtend  phoenix表扩展字段
     */
    private void checkTable(String sinkTable, String sinkColumns, String sinkPk, String sinkExtend) {
        PreparedStatement preparedStatement = null;
        try {
            //处理特殊字段
            if (sinkPk == null || "".equals(sinkPk)) {
                sinkPk = "id";
            }
            if (sinkExtend == null) {
                sinkExtend = "";
            }
            //拼接sql: create table if not exits db.tn(id varchar primary key,bb varchar,cc varchar)xxx
            StringBuilder createTableSql = new StringBuilder("create table if not exists ")
                    .append(GmallConfig.HBASE_SCHEMA)
                    .append(".")
                    .append(sinkTable)
                    .append("(");
            String[] columns = sinkColumns.split(",");
            for (int i = 0; i < columns.length; i++) {
                String column = columns[i];
                //            判断是否是主键
                if (sinkPk.equals(column)) {
                    createTableSql.append(column).append(" varchar primary key");
                } else {
                    createTableSql.append(column).append(" varchar");
                }
                //            判断是否是最后一个字段
                if (i < columns.length - 1) {
                    createTableSql.append(",");
                }
            }

            createTableSql.append(")").append(sinkExtend);
            //编译sql
            System.out.println("建表语句为：" + createTableSql);
//            编译时异常，不处理运行不了
            preparedStatement = connection.prepareStatement(createTableSql.toString());

            //执行sql，建表
            preparedStatement.execute();
        } catch (SQLException e) {
            e.printStackTrace();
//            将编译时异常转化为运行时异常主动抛出停止程序
            throw new RuntimeException("建表失败：" + sinkTable);
        } finally {
            //释放资源
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException throwables) {
                    throwables.printStackTrace();
                }
            }
        }
    }

    //    {"database":"gmall","table":"base_trademark","type":"bootstrap-insert","ts":1673323672,"data":{"id":1,"tm_name":"三星","logo_url":"/static/default.jpg"}}
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
//        1、获取广播的配置流数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String table = value.getString("table");
        TableProcess tableProcess = broadcastState.get(table);

        if (tableProcess != null) {
            //        2、过滤主流数据和字段
            filterColumn(value.getJSONObject("data"), tableProcess.getSinkColumns());
//            3、补充sinkTable并写出到流
            value.put("sinkTable", tableProcess.getSinkTable());
            out.collect(value);

        } else {
            System.out.println("找不到对应的key：" + table);
        }

    }

    /**
     * 过滤主流字段
     *
     * @param data        {"id":1,"tm_name":"三星","logo_url":"/static/default.jpg"}
     * @param sinkColumns "id,tm_name"
     */
    private void filterColumn(JSONObject data, String sinkColumns) {

        String[] columns = sinkColumns.split(",");
        List<String> columnList = Arrays.asList(columns);

        Set<Map.Entry<String, Object>> entries = data.entrySet();
//        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
//        while (iterator.hasNext()){
//            Map.Entry<String, Object> next = iterator.next();
//            if (!columnList.contains(next.getKey())){
//                iterator.remove();
//            }
//        }

        entries.removeIf(next -> !columnList.contains(next.getKey()));
    }

    @Override
    public void close() throws Exception {
        if (connection != null) {
            connection.close();
        }
    }

}
